package com.zlx.rulemk.marketing.utils;

import com.typesafe.config.Config;
import com.zlx.rulemk.marketing.constant.ConfigNames;
import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer;

import java.util.Properties;

/**
 * @Author: zhanglingxing
 * @CreateTime: 2022-07-08
 * @Description: TODO
 * @Version: 1.0
 */
public class KafkaSourceBuilderUtils {

    public static FlinkKafkaConsumer<String> build(Config config,String topic, String groupID){
        Properties props = new Properties();
        props.setProperty("bootstrap.servers",config.getString(ConfigNames.KAFKA_BOOTSTRAP_SERVERS));
        props.setProperty("auto.offset.reset", config.getString(ConfigNames.KAFKA_OFFSET_AUTORESET));
        props.setProperty("group.id", groupID);
        FlinkKafkaConsumer<String> kafkaConsumer = new FlinkKafkaConsumer<>(topic, new SimpleStringSchema(), props);
        return kafkaConsumer;
    }


}
